package events

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	logger "gitee.com/youkelike/zlog"
	"gitee.com/youkelike/zmigrate/migrator"
	"gitee.com/youkelike/zmigrate/migrator/fixer"
	"gitee.com/youkelike/zmigrate/pkg/saramax"
	"github.com/IBM/sarama"
	"gorm.io/gorm"
)

// 用于修复数据的消费者，
// 为每个需要迁移的表构造一对生产者-消费者，根据表名构造 topic，
// 全量、增量修复共用
type FixDataConsumer[T migrator.Entity] struct {
	client   sarama.Client
	l        logger.Logger
	srcFirst *fixer.OverrideFixer[T]
	dstFirst *fixer.OverrideFixer[T]
	topic    string
}

func NewFixDataConsumer[T migrator.Entity](
	client sarama.Client,
	l logger.Logger,
	src *gorm.DB,
	dst *gorm.DB) (*FixDataConsumer[T], error) {
	srcFirst, err := fixer.NewOverrideFixer[T](src, dst)
	if err != nil {
		return nil, err
	}
	dstFirst, err := fixer.NewOverrideFixer[T](dst, src)
	if err != nil {
		return nil, err
	}

	var t T
	topic := "migrator_" + t.TableName()
	return &FixDataConsumer[T]{
		client:   client,
		l:        l,
		srcFirst: srcFirst,
		dstFirst: dstFirst,
		topic:    topic,
	}, nil
}

// Start 这边就是自己启动 goroutine 了
func (r *FixDataConsumer[T]) Start() error {
	group := "migrator-fix"
	cg, err := sarama.NewConsumerGroupFromClient(group, r.client)
	if err != nil {
		return err
	}
	go func() {
		r.l.Debugf("启动消费循环，topic=%s，group=%s\n", r.topic, group)
		err := cg.Consume(context.Background(),
			[]string{r.topic},
			saramax.NewHandler[InconsistentEvent](r.l, r.Consume))
		if err != nil {
			var t T
			r.l.Errorw("退出了消费循环异常",
				"err", err,
				"FixDataConsumer", t.TableName())
		}
	}()
	return err
}

func (r *FixDataConsumer[T]) Consume(msg *sarama.ConsumerMessage, e InconsistentEvent) (err error) {
	r.l.Debugw("数据修复", "InconsistentEvent", e)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	switch e.Direction {
	case "SRC":
		err = r.srcFirst.Fix(ctx, e.ID)
	case "DST":
		err = r.dstFirst.Fix(ctx, e.ID)
	default:
		err = errors.New("未知的校验方向")
	}
	if err != nil {
		var t T
		data, _ := json.Marshal(e)
		r.l.Errorw("修复数据报错",
			"err", err,
			"FixDataConsumer", t.TableName(),
			"data", string(data),
		)
	}
	return err
}
